package com.idanchuang.component.redis.util;

import com.google.common.collect.Lists;
import com.idanchuang.component.base.exception.common.ErrorCode;
import com.idanchuang.component.base.exception.core.ExFactory;
import com.idanchuang.component.core.util.SpringUtil;
import com.idanchuang.component.logback.metric.MetricLogger;
import com.idanchuang.component.redis.util.pubsub.LockSemaphore;
import com.idanchuang.component.redis.util.pubsub.LockReleasePubSubs;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import static com.idanchuang.component.redis.config.DRedisLockConfig.*;

/**
 * 基于Redis的分布式锁(线程内可重入)
 * @author yjy
 * @date 2019/11/27 11:07
 **/
public class DRedisLock implements Lock {

    private static final Logger log = LoggerFactory.getLogger(DRedisLock.class);

    public static final String HASH_TAG = "DRedisLock";

    /** 序列值, 用于确保锁value的唯一性 */
    private static AtomicLong SERIAL_NUM;
    /** 最大序列值 */
    private static long MAX_SERIAL;
    /** 本机host */
    private static String CURRENT_HOST;
    /** 用于自动续期的定时任务时间轮 */
    private static final HashedWheelTimer HASHED_WHEEL_TIMER;

    /** StringRedisTemplate */
    private final StringRedisTemplate redisTemplate;
    /** 锁Key */
    private final String lockKey;
    /** 锁资源 */
    private final String lockName;
    /** 锁超时时间(单位毫秒) */
    private final long timeout;
    /** 主机+线程id */
    private final String hostThreadId;
    /** 锁定值前缀, 用来判断是否可重入 */
    private final String lockValuePrefix;
    /** 锁定值 */
    private final String lockValue;
    /** 是否自动续期 */
    private final boolean autoRenew;
    /** 当前的锁持有人有可能是当前线程 */
    private boolean maybeSameThread = true;
    /** 是否重入 */
    private boolean reentrant = false;
    /** 是否持有锁 */
    private boolean locked = false;
    /** 锁定开始时间 */
    private long lockStartTime;
    /** 当前的自动续期任务 */
    private Timeout renewTask;

    private final LockSemaphore lockSemaphore;

    static {
        // 每100ms走一格, 一圈一共1024格(必须是2的次方), 相当于 102.4s走一圈
        HASHED_WHEEL_TIMER = new HashedWheelTimer(Executors.defaultThreadFactory(),
                100, TimeUnit.MILLISECONDS, 1024, false);
        try {
            SERIAL_NUM = new AtomicLong(0);
            MAX_SERIAL = 99999999L;
            CURRENT_HOST = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            CURRENT_HOST = UUID.randomUUID().toString();
            log.warn("DRedisLock > can not get local host, use uuid: {}", CURRENT_HOST);
        }
    }

    public DRedisLock(String lockName) {
        this(lockName, DEFAULT_TIMEOUT, false);
    }

    public DRedisLock(String lockName, boolean autoRenew) {
        this(lockName, DEFAULT_TIMEOUT, autoRenew);
    }

    public DRedisLock(String lockName, long timeout) {
        this(lockName, timeout, false);
    }

    @Deprecated
    public DRedisLock(String lockName, long timeout, long loopInterval) {
        this(lockName, timeout, false);
    }

    public DRedisLock(String lockName, long timeout, boolean autoRenew) {
        if (lockName == null) {
            throw ExFactory.throwSystem("lockName must assigned");
        }
        this.redisTemplate = SpringUtil.getBean(StringRedisTemplate.class);
        this.lockName = lockName;
        this.lockKey = LOCK_PREFIX + lockName + "{" + HASH_TAG + "}";
        this.timeout = timeout;
        this.hostThreadId = CURRENT_HOST + ":" + Thread.currentThread().getId();
        this.lockValuePrefix = this.hostThreadId + ":";
        this.lockValue = this.lockValuePrefix + getNextSerial();
        this.autoRenew = autoRenew;
        this.lockSemaphore = new LockSemaphore(this.lockKey, 0);
    }

    /**
     * 获取锁, 如果锁被持有, 将一直等待, 直到超出默认的的DEFAULT_TRY_LOCK_TIMEOUT
     */
    @Override
    public void lock() {
        try {
            if (!tryLock(DEFAULT_TRY_LOCK_TIMEOUT, TimeUnit.MILLISECONDS)) {
                throw ExFactory.throwWith(ErrorCode.CONFLICT, "try lock timeout, lockKey: " + this.lockKey);
            }
        } catch (InterruptedException e) {
            throw ExFactory.throwWith(ErrorCode.CONFLICT, e.getMessage());
        }
    }

    /**
     * 尝试获取锁, 如果锁被持有, 则等待相应的时间(等待锁时可被中断)
     * @throws InterruptedException 被中断等待
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        if (!tryLock(DEFAULT_TRY_LOCK_TIMEOUT, true)) {
            throw ExFactory.throwWith(ErrorCode.CONFLICT, "try lock timeout, lockKey: " + this.lockKey);
        }
    }

    /**
     * 尝试获取锁, 只会立即获取一次, 如果锁被占用, 则返回false, 获取成功则返回true
     * @return 是否成功获取锁
     */
    @Override
    public boolean tryLock() {
        long ttl = tryLock0();
        return ttl < 0;
    }

    private Long tryLock0() {
        this.lockStartTime = 0L;
        try {
            // 尝试抢占锁
            String script = "if redis.call('exists', KEYS[1])==0 " +
                                "then redis.call('set', KEYS[1], ARGV[1]); " +
                                "redis.call('pexpire', KEYS[1], ARGV[3]); " +
                                // 占用成功, 返回-1
                                "return -1; " +
                            "end; " +
                            // 检查重入的可能性(锁值为当前线程表示可重入)
                            "if ARGV[4] " +
                                "then local curValue = redis.call('get', KEYS[1]); " +
                                "if string.find(curValue, ARGV[2])==1 " +
                                    // 重入
                                    "then local curTtl = redis.call('pttl', KEYS[1]); " +
                                    "redis.call('pexpire', KEYS[1], curTtl + ARGV[3]); " +
                                    // 重入成功, 返回-2
                                    "return -2; " +
                                "end " +
                            "end " +
                            // 占用失败, 返回当前锁过期时间
                            "return redis.call('pttl', KEYS[1]); ";
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setResultType(Long.class);
            redisScript.setScriptText(script);
            List<String> keys = new ArrayList<>();
            keys.add(this.lockKey);
            Long ttl = redisTemplate.execute(redisScript, keys,
                    this.lockValue,
                    this.lockValuePrefix,
                    String.valueOf(this.timeout),
                    String.valueOf(this.maybeSameThread)
            );
            if (ttl < 0) {
                // 处理成功
                handleLockSuccess(ttl == -2);
            } else if (this.maybeSameThread) {
                // 看来当前锁持有人并不是自己这个线程
                maybeSameThread = false;
            }
            return ttl;
        } catch (Exception e) {
            log.error("tryLock error, do unlock, lockKey: {}, lockValue: {}", this.lockKey, lockValue, e);
            unlock();
            return 0L;
        }
    }

    /**
     * 处理获取锁成功
     * @param reentrant 是否重入
     */
    private void handleLockSuccess(boolean reentrant) {
        this.reentrant = reentrant;
        this.locked = true;
        this.lockStartTime = System.currentTimeMillis();
        if (this.autoRenew) {
            // 自动续期
            this.registerAutoRenew();
        }
        log.debug("Lock success, reentrant: {}, lockKey: {}, lockValue: {}",
                reentrant, this.lockKey, this.lockValue);
    }

    /**
     * 尝试获取锁, 如果锁被占用, 则持续尝试获取, 直到超过指定的time时间
     * @param time 等待锁的时间
     * @param unit time的单位
     * @return 是否成功获取锁
     * @throws InterruptedException 被中断
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return tryLock(unit.toMillis(time), false);
    }

    /**
     * 尝试获取锁, 如果锁被占用, 则持续尝试获取, 直到超过指定的time时间
     * @param waitTime 等待锁的时间, 单位: 毫秒
     * @param interruptibly 等待是否可被中断
     * @return 是否成功获取锁
     * @throws InterruptedException 被中断
     */
    private boolean tryLock(long waitTime, boolean interruptibly) throws InterruptedException {
        if (tryLock()) {
            return true;
        }
        try {
            long finalTime = waitTime + System.currentTimeMillis();
            LockReleasePubSubs.register(lockSemaphore);
            while (true) {
                long ttl = tryLock0();
                if (ttl < 0) {
                    return true;
                }
                long now = System.currentTimeMillis();
                long remainTime = finalTime - now;
                if (remainTime < 0) {
                    return false;
                }
                try {
                    lockSemaphore.tryAcquire(Math.min(remainTime, ttl), TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    if (interruptibly) {
                        throw e;
                    }
                }
            }
        } finally {
            LockReleasePubSubs.unregister(lockSemaphore);
        }
    }

    /**
     * 释放锁
     */
    @Override
    public void unlock() {
        try {
            if (!locked) {
                return;
            }
            if (this.reentrant) {
                handleUnlockSuccess(true);
                return;
            }
            // 使用lua脚本处理锁判断和释放
            String script = "if redis.call('get', KEYS[1])==ARGV[1] " +
                                "then " +
                                    "redis.call('del', KEYS[1]) " +
                                    "redis.call('publish', KEYS[2], ARGV[2]); " +
                                    "return true " +
                                "else " +
                                    "return false " +
                                "end";
            DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>();
            redisScript.setResultType(Boolean.class);
            redisScript.setScriptText(script);
            Boolean res = this.redisTemplate.execute(redisScript,
                    Lists.newArrayList(this.lockKey, LockReleasePubSubs.CHANNEL),
                    this.lockValue, this.lockKey);
            if (res != null && res) {
                handleUnlockSuccess(false);
                return;
            }
            log.warn("Unlock failed, lockKey: {}, lockValue: {}", this.lockKey, this.lockValue);
        } catch (Exception e) {
            log.error("Unlock error", e);
        } finally {
            LockReleasePubSubs.ASYNC.execute(() -> {
                Map<String, Object> meta = new HashMap<>(4);
                meta.put("lockName", this.lockName);
                boolean success = this.lockStartTime > 0;
                meta.put("success", success);
                long rt = success ? (System.currentTimeMillis() - this.lockStartTime) : 0;
                MetricLogger.log("RedisLock", rt, meta);
            });
        }
    }

    private void handleUnlockSuccess(boolean reentrant) {
        this.locked = false;
        if (this.autoRenew) {
            // 关闭自动续期任务
            cancelAutoRenew();
        }
        log.debug("Unlock success, reentrant: {}, lockKey: {}, lockValue: {}",
                reentrant, this.lockKey, this.lockValue);
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }

    /**
     * 注册自动续期
     */
    private void registerAutoRenew() {
        LockReleasePubSubs.ASYNC.execute(() -> {
            this.renewTask = HASHED_WHEEL_TIMER.newTimeout(to -> doRenew(),
                    this.timeout / 3, TimeUnit.MILLISECONDS);
        });
    }

    /**
     * 取消自动续期
     */
    private void cancelAutoRenew() {
        LockReleasePubSubs.ASYNC.execute(() -> {
            if (this.renewTask != null) {
                this.renewTask.cancel();
            }
        });
    }

    private void doRenew() {
        String script = "local val,ttl=ARGV[1],ARGV[2] ";
        script += "if redis.call('EXISTS', KEYS[1])==1 " +
                "then local curValue = redis.call('GET', KEYS[1]) " +
                "if curValue==val " +
                "then redis.call('PEXPIRE', KEYS[1], ttl) " +
                "return true " +
                "else " +
                "return false " +
                "end " +
                "else " +
                "return false " +
                "end ";
        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>();
        redisScript.setResultType(Boolean.class);
        redisScript.setScriptText(script);
        List<String> keys = new ArrayList<>();
        keys.add(this.lockKey);
        long expire = Math.max(this.timeout, 1000L);
        Boolean success = redisTemplate.execute(redisScript, keys, this.lockValue, String.valueOf(expire));
        if (success != null && success) {
            this.registerAutoRenew();
        }
    }

    /**
     * @return 下一个序列值
     */
    private static synchronized long getNextSerial() {
        long serial = SERIAL_NUM.incrementAndGet();
        if (serial > MAX_SERIAL) {
            serial = serial - MAX_SERIAL;
            SERIAL_NUM.set(serial);
        }
        return serial;
    }

    public static AtomicLong getSerialNum() {
        return SERIAL_NUM;
    }

    public static long getMaxSerial() {
        return MAX_SERIAL;
    }

    public static String getCurrentHost() {
        return CURRENT_HOST;
    }

    public String getLockKey() {
        return lockKey;
    }

    public long getTimeout() {
        return timeout;
    }

    public String getHostThreadId() {
        return hostThreadId;
    }

    public String getLockValue() {
        return lockValue;
    }

    public boolean isReentrant() {
        return reentrant;
    }

    @Override
    public String toString() {
        return "DRedisLock{" +
                "lockKey='" + lockKey + '\'' +
                ", timeout=" + timeout +
                ", hostThreadId='" + hostThreadId + '\'' +
                ", lockValue='" + lockValue + '\'' +
                ", reentrant=" + reentrant +
                '}';
    }
}
